package com.example.mq.mqserver.dataCenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 内存数据管理
 * 使用此类统一管理内存之中的数据
 * 存在多线程并发的情况 所以需要注意存储的结构
 */
public class MemoryDataCenter {
    // 存储交换机的数据信息 Key --- exchangeName、Value --- Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // 存储队列的数据消息 Key --- queueName、Value --- MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 存储绑定关系的数据消息 第一个Key --- exchangeName、Value --- Map --- 第二个Key --- queueName,Value --- Binding对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 存储消息的数据 key ---- messageId, value --- Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // 存储队列和消息所属关系的数据 Key --- queueName、Value --- LinkedList --- Message对象
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap  = new ConcurrentHashMap<>();
    // 存储未处理的消息的数据
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 添加新交换机成功. exchangeName=" + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 删除交换机成功.exchangeName=" + exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 添加新队列成功. queueName=" + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 删除队列成功. queueName=" + queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
        //先根据exchangeName 交换机是否有绑定关系
        ConcurrentHashMap<String,Binding> curBindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        synchronized (curBindingMap){
            //再根据queueName 查看绑定是否存在,如果存在抛出异常 不存在才进行插入
            if(curBindingMap.get(binding.getQueueName()) != null){
                throw new MqException("[MemoryDataCenter] 已存在绑定关系. exchangeName = "+binding.getExchangeName()
                        +",queueName"+binding.getQueueName());
            }
            //没有绑定关系 此时将绑定插入到 bindingsMap 中
            curBindingMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功. exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    //获取绑定关系:
    //1.获取某个 交换机 和 某个队列之间的绑定关系: exchangeName + queueName
    //2.获取某个 交换机 的所有绑定关系 :exchangeName
    public Binding getBinding(String exchangeName,String queueName){
        //先查看 交换机 与 队列 之间的绑定是否存在
        ConcurrentHashMap<String,Binding> curBindingMap = bindingsMap.get(exchangeName);
        if(curBindingMap == null){
            return null;
        }
        return curBindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    //获取某个交换机的所有绑定
    public ConcurrentHashMap<String,Binding> getAllBinding(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    //删除某个绑定关系
    public void deleteBinding(Binding binding) throws MqException {
        //查看当前是否存在该绑定关系
        ConcurrentHashMap<String,Binding> curBindingMap = bindingsMap.get(binding.getExchangeName());
        if(curBindingMap == null){
            throw new MqException("[MemoryDataCenter] 绑定关系不存在. exchangeName = "+binding.getExchangeName()
            +",queueName =" + binding.getQueueName());
        }
        //存在就移除该绑定关系
        curBindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 删除绑定关系成功. exchangeName = "+binding.getExchangeName()
                +",queueName =" + binding.getQueueName());
    }

    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 添加新消息成功. messageId=" + message.getMessageId());
    }

    // 根据 id 查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 根据 id 删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除. messageId=" + messageId);
    }

    //发送消息到指定队列之中
    public void sendMessage(MSGQueue queue,Message message){
        //先查看该队列存不存在消息链表
        LinkedList<Message> messageList = queueMessageMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        //往链表里面插入该条消息
        //LinkedList 非线程安全.需要加锁
        synchronized (messageList){
            messageList.add(message);
        }
        //把消息加入到消息中心
        //此处重复插入也没有关系 因为消息只跟消息ID进行了唯一关联,并不会有其他影响.
        addMessage(message);
        System.out.println("[MemoryDataCenter] 成功投递消息入队列. messageId = "+message.getMessageId()+
                ", queueName = "+queue.getName());
    }

    //获取队列之中的消息
    public Message pollMessage(String queueName){
        //获取消息链表
        LinkedList<Message> messageList = queueMessageMap.get(queueName);
        //链表不存在 返回null
        if(messageList == null){
            return null;
        }

        synchronized (messageList){
            //如果链表为空 也返回null
            if(messageList.size() == 0){
                return  null;
            }
            //采用头删 返回消息
            Message curMessage = messageList.remove(0);
            System.out.println("[MemoryDataCenter] 消息成功从队列中取出. messageId = "+curMessage.getMessageId()+
                    ", queueName = "+queueName);
            return curMessage;
        }
    }

    //获取队列之中的消息个数
    public int getMessageCount(String queueName){
        //获取消息链表
        LinkedList<Message> messageList = queueMessageMap.get(queueName);
        if(messageList == null){
            return 0;
        }
        synchronized (messageList){
            return messageList.size();
        }
    }

    //添加未确认的消息进队列之中
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列 queueName = "+queueName
                +", messageId = "+ message.getMessageId());
    }

    //消息已得到确认 删除这个未确认的消息
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageMap = queueMessageWaitAckMap.get(queueName);
        if(messageId == null){
            return;
        }
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除成功. queueName = "+queueName
                +",messageId = "+messageId);
    }

    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageMap = queueMessageWaitAckMap.get(queueName);
        if(messageId == null){
            return null;
        }
        return messageMap.get(messageId);
    }

    //读取硬盘上的数据 硬盘中之前持久化存储的各个维度的数据都恢复到内存中.
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //先清空所有数据结构的内存消息
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();

        //获取硬盘上的交换机数据
        List<Exchange> exchangeList = diskDataCenter.selectAllExchanges();
        for(Exchange exchange : exchangeList){
            exchangeMap.put(exchange.getName(),exchange);
        }

        //获取硬盘上的队列数据
        List<MSGQueue> queueList = diskDataCenter.selectAllQueues();
        for(MSGQueue queue : queueList){
            queueMap.put(queue.getName(),queue);
        }

        //获取所有的绑定消息
        List<Binding> bindingList = diskDataCenter.selectAllBindings();
        for(Binding binding : bindingList){
            ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }

        //遍历所有队列 获取所有的消息
        for(MSGQueue msgQueue : queueList){
            LinkedList<Message> messageList = diskDataCenter.loadAllMessageFromQueue(msgQueue.getName());
            //将消息链表添加进队列之中
            queueMessageMap.put(msgQueue.getName(),messageList);
            //将每个消息添加进消息中心
            for(Message message : messageList){
                messageMap.put(message.getMessageId(),message);
            }
        }

        // 针对 "未确认的消息" 这部分内存中的数据, 不需要从硬盘恢复.
        // 一旦在等待 ack 的过程中, 服务器重启了, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" .
    }

}
